package com.shujia.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
 * 重分区
 *
 * repartition
 *
 * coalesce
 *
 * 面试题：如何修改rdd中的分区，区别是什么？
 *
 */
object Demo4Coalesce {
  def main(args: Array[String]): Unit = {
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("cache")
      //      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    val sparkContext: SparkContext = sparkSession.sparkContext

    val studentsRDD: RDD[String] = sparkContext.textFile("spark/data/students.csv") // 1000条数据

    println(s"studentsRDD的分区数量为：${studentsRDD.getNumPartitions}") //一开始RDD中的分区数量取决于block块的数量

    /**
     * repartition：对RDD进行重分区，返回一个新的RDD
     * repartition可以增加RDD的分区数量或者减少RDD的分区数量
     *
     * 增加或者减少分区是根据资源而定
     * 若资源充足，可以适当的增加分区，提高task任务数量并行度，加快整个spark作业执行
     *
     * rdd中一个分区数据是否对应有且仅有后一个分区数据，如果前一个RDD的分区数据对应的是后一个RDD中多个分区，那么就是shuffle
     * 使用repartition增加分区会产生shuffle,数据重构,不是按照原来的顺序
     * 减少分区也会产生shuffle
     *
     */
    //    val students2RDD: RDD[String] = studentsRDD.repartition(10)
    ////    students2RDD.foreach(println)
    //    println(s"students2RDD的分区数量为：${students2RDD.getNumPartitions}")
    //
    //    val students3RDD: RDD[String] = students2RDD.repartition(1)
    //    println(s"students3RDD的分区数量为：${students3RDD.getNumPartitions}")
    //    students3RDD.foreach(println)


    /**
     * coalesce 重分区，增加分区和减少分区
     *
     * 传入两个值：numPartitions: Int 目标分区数量, shuffle: Boolean = false 是否产生shuffle
     *
     * coalesce增加分区，需要产生shuffle
     * coalesce减少分区，可以不用shuffle，也可以用。使用场景，如果要合并小文件的话，可以使用coalesce且不产生shuffle
     */
    val studentsRDD2: RDD[String] = studentsRDD.coalesce(10, shuffle = true)
    println(s"studentsRDD2的分区数量为：${studentsRDD2.getNumPartitions}")
    studentsRDD2.foreach(println)

    val studentRDD3: RDD[String] = studentsRDD2.coalesce(2, shuffle = false)
    println(s"studentRDD3的分区数量为：${studentRDD3.getNumPartitions}")
    studentRDD3.foreach(println)

    while (true) {

    }


  }
}
